package io.github.wanggit.antrpc.client.zk.zknode;

import io.github.wanggit.antrpc.commons.bean.SubscribeNode;
import io.github.wanggit.antrpc.commons.constants.ConstantValues;
import io.github.wanggit.antrpc.commons.utils.JsonUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.CreateMode;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@Slf4j
public class ReportSubscribe implements IReportSubscriber, Runnable {

    private final IZkNodeOperator zkNodeOperator;
    private final Set<String> subscribePaths = new HashSet<>();
    private final Set<String> subscribeInterfaceName = new HashSet<>();
    private final Set<SubscribeNode> subscribeNodes = new HashSet<>();
    private final ScheduledExecutorService executorService;
    private final INodeHostContainer nodeHostContainer;

    public ReportSubscribe(IZkNodeOperator zkNodeOperator, INodeHostContainer nodeHostContainer) {
        this.zkNodeOperator = zkNodeOperator;
        this.nodeHostContainer = nodeHostContainer;
        executorService = Executors.newSingleThreadScheduledExecutor();
        executorService.scheduleAtFixedRate(this, 60, 60, TimeUnit.SECONDS);
    }

    @Override
    public boolean wasSubscribed(String interfaceName) {
        return subscribeInterfaceName.contains(interfaceName);
    }

    @Override
    public void report(SubscribeNode subscribeNode) {
        subscribeNodes.add(subscribeNode);
        internalDoReport(subscribeNode);
    }

    @Override
    public void run() {
        if (log.isDebugEnabled()) {
            log.debug(
                    "Periodically check whether the reported subscription node. subscribeNodes="
                            + JsonUtils.toJsonPrettyString(subscribeNodes));
        }
        subscribeNodes.forEach(this::internalDoReport);
        if (null != nodeHostContainer) {
            // 本地节点服务列表缓存中是否存在
            nodeHostContainer.checkLocalCacheNodes(subscribeNodes);
        }
    }

    @Override
    public void unReport() {
        if (log.isInfoEnabled()) {
            log.info("All subscribe will be cancel.");
        }
        subscribeNodes.clear();
        for (String path : subscribePaths) {
            zkNodeOperator.deleteNode(path);
        }
        subscribePaths.clear();
        subscribeInterfaceName.clear();
        if (null != executorService && !executorService.isShutdown()) {
            executorService.shutdown();
        }
    }

    private void internalDoReport(SubscribeNode subscribeNode) {
        String fullPath = getFullpath(subscribeNode);
        // 远程节点是否存在
        if (!zkNodeOperator.existsNode(fullPath)) {
            byte[] bytes = JsonUtils.toJsonBytes(subscribeNode);
            zkNodeOperator.remoteCreateZkNode(fullPath, bytes, CreateMode.EPHEMERAL);
            subscribePaths.add(fullPath);
            subscribeInterfaceName.add(subscribeNode.getClassName());
        }
    }

    private String getFullpath(SubscribeNode subscribeNode) {
        return "/"
                + ConstantValues.ZK_ROOT_SUBSCRIBE_NODE_NAME
                + "/"
                + subscribeNode.getHost()
                + "/"
                + subscribeNode.getClassName();
    }
}
